#include <time.h>

#include <list.h>
#include "core.h"
#include "schedule.h"
#include "volume_proto.h"

#include "lsv_help.h"
#include "lsv_log.h"
#include "lsv_gc_internal.h"

#define VALUEUP_DIRECTLY 1

int lsv_gc_valueup(lsv_volume_proto_t *lsv_info, lsv_u32_t chunk_id, lsv_u32_t chunk_page_off) {
        int ret = 0;

        if (lsv_feature & LSV_FEATURE_GC) {
#if VALUEUP_DIRECTLY
                ret = lsv_gc_do_valueup(lsv_info, chunk_id, chunk_page_off);
                if (unlikely(ret)) {
                        GOTO(ERR0, ret);
                }
#else
                lsv_log_info_t *log_info = NULL;
                lsv_loggc_valueup_queue_t *valueup_queue = NULL;
                lsv_log_gc_arg_t * gc_arg = NULL;
                log_info = lsv_info->log_info;
                valueup_queue = &log_info->gc_context->valueup_queue;

                // @malloc ERR1
                ret = ymalloc((void **)&gc_arg, sizeof(lsv_log_gc_arg_t));
                if (unlikely(ret)) {
                        ret = -ENOMEM;
                        DERROR("malloc gc_arg failed, errno:%d\n", ret);
                        GOTO(ERR0, ret);
                }

                gc_arg->msg_type = LSV_LOGGC_ARG_TPYE_VALUEUP;
                gc_arg->chunk_id = chunk_id;
                gc_arg->chunk_page_off = chunk_page_off;
                gc_arg->is_free = 1;

                // lsv_lock(&valueup_queue->lock);
                list_add_tail(&gc_arg->hook, &valueup_queue->queue);
                valueup_queue->count++;
                // lsv_unlock(&valueup_queue->lock);

                DINFO_NOP("offer a valueup, chunk_id:%u chunk_offset:%u gc_queue_count:%u\n",
                          chunk_id,
                          chunk_page_off,
                          valueup_queue->count);
#endif
        }

        return 0;
ERR0:
        return ret;
}

int lsv_gc_valueup_commit(lsv_volume_proto_t *lsv_info) {
        lsv_log_info_t *log_info = NULL;
        lsv_gc_valueup_queue_t *valueup_queue = NULL;

        log_info = lsv_info->log_info;
        valueup_queue = &log_info->gc_context->valueup_queue;

#if VALUEUP_DIRECTLY
        YASSERT(valueup_queue->count == 0);
#else
        DINFO_NOP("commit gc valueup, %u\n", valueup_queue->count);
        if (lsv_feature & LSV_FEATURE_GC) {
                lsv_log_gc_arg_t * gc_arg = NULL;
                struct list_head *pos, *n;

                list_for_each_safe(pos, n, &valueup_queue->queue) {
                        gc_arg = (lsv_log_gc_arg_t *) pos;

                        // TODO 直接处理，还是加入队列？
                        lsv_log_gc_offer(lsv_info, gc_arg);

                        list_del_init(pos);
                        valueup_queue->count--;
                }
        }
#endif
        return 0;
}

int lsv_gc_do_valueup(lsv_volume_proto_t *lsv_info, lsv_u32_t chunk_id, lsv_u32_t chunk_page_off) {
        int ret = 0;
        lsv_log_info_t *log_info = NULL;
        lsv_gc_context_t *gc_context = NULL;
        lsv_gc_heap_t *gc_heap = NULL;
        lsv_gc_logctrl_t *logctrl = NULL;

        log_info = lsv_info->log_info;
        gc_context = log_info->gc_context;
        gc_heap = &gc_context->gc_heap;

        logctrl = (lsv_gc_logctrl_t*) hash_table_find(gc_context->logctrl_tab, &chunk_id);
        if (NULL == logctrl) {
                ret = ENOKEY;
                DINFO_NOP("no that log, may be in old_storage. chunk_id:%u,chunk_page_off:%u,all_gc_value:%u\n",
                          chunk_id,
                          chunk_page_off,
                          gc_context->old_storage.all_gc_value);
                if (gc_context->old_storage.all_gc_value < 0xFFFFFFFF) {
                        gc_context->old_storage.all_gc_value++;
                }
        } else {
                DINFO("do valueup, chunk_id %u chunk_page_off %u\n", chunk_id, chunk_page_off);
                lsv_gc_heap_valueup(gc_heap, logctrl);

                // lsv_lock(&gc_heap->lock); //protect heap to storage
                // must check again,logctrl may be free in get lock that make a yield
                // logctrl = (lsv_loggc_logctrl_t*) hash_table_find(gc_context->logctrl_tab, &chunk_id);
                // if (NULL != logctrl) {
                // }
                //  lsv_unlock(&gc_heap->lock);
        }

        return ret;
}
